package olap4cloud.hadop.statistic;

import java.io.IOException;

import olap4cloud.data.DataStruct;
import olap4cloud.util2.MyDateUtil;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class JobStatistic extends Configuration implements Tool {

	/**
	 * @uml.property  name="conf"
	 * @uml.associationEnd  
	 */
	private Configuration conf;

	/**
	 * @return
	 * @uml.property  name="conf"
	 */
	@Override
	public Configuration getConf() {
		// TODO Auto-generated method stub
		return conf;
	}

	/**
	 * @param arg0
	 * @uml.property  name="conf"
	 */
	@Override
	public void setConf(Configuration arg0) {
		// TODO Auto-generated method stub
		this.conf = arg0;
	}

	/**
	 * @author  LuyenThanhDat
	 */
	public static class MapperClass extends
			Mapper<LongWritable, Text, Model, IntWritable> {

		/**
		 * @uml.property  name="keyOutPut"
		 * @uml.associationEnd  
		 */
		Model keyOutPut = new Model();
		long hourRoundDown = 0;
		IntWritable valoueOut = new IntWritable(1);

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			DataStruct dataStruct = new DataStruct(value.toString());
			hourRoundDown = MyDateUtil.roundDownHourTime(dataStruct.getTimestamp());
			
			System.out.println(hourRoundDown);
			
			keyOutPut.setKey((hourRoundDown+"").getBytes());
			keyOutPut.setColumF(StatisticsTable.columfHour);

			keyOutPut.setColumQ(Bytes.toBytes("d_" + dataStruct.getDomain()));
			context.write(keyOutPut, valoueOut);

			keyOutPut
					.setColumQ(Bytes.toBytes("c_" + dataStruct.getChannelId()));
			context.write(keyOutPut, valoueOut);

			keyOutPut.setColumQ(Bytes.toBytes("b_" + dataStruct.getBrowser()));
			context.write(keyOutPut, valoueOut);

			keyOutPut.setColumQ(Bytes.toBytes("total"));
			context.write(keyOutPut, valoueOut);

			keyOutPut.setKey(Bytes.add(Bytes.toBytes(hourRoundDown+""),
					Bytes.toBytes(dataStruct.getDomain())));

			keyOutPut.setColumQ(Bytes.toBytes("b_" + dataStruct.getBrowser()));
			context.write(keyOutPut, valoueOut);

			keyOutPut
					.setColumQ(Bytes.toBytes("c_" + dataStruct.getChannelId()));
			context.write(keyOutPut, valoueOut);

			keyOutPut
					.setColumQ(Bytes.toBytes("c_" + dataStruct.getChannelId()));
			context.write(keyOutPut, valoueOut);

			keyOutPut
					.setColumQ(Bytes.toBytes("t_" + dataStruct.getChannelId()));
			context.write(keyOutPut, valoueOut);

			keyOutPut
					.setColumQ(Bytes.toBytes("total" + dataStruct.getChannelId()));
			context.write(keyOutPut, valoueOut);
		}
	}

	public static class ReducerClass extends
			Reducer<Model, IntWritable, Model, LongWritable> {

		@Override
		public void reduce(Model key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			long pageview = 0;
			for (IntWritable v : values) {
				pageview = pageview + 1;
			}
			context.write(key, new LongWritable(pageview));
		}
	}

	// public static class TableReducerClass extends
	// TableReducer<Model, IntWritable, ImmutableBytesWritable> {
	//
	// @Override
	// public void reduce(Model key, Iterable<IntWritable> values,
	// Context context) throws IOException, InterruptedException {
	// long pageview = 0;
	// for (IntWritable v : values) {
	// pageview = pageview + 1;
	// }
	//
	// //Put p = new Put(key.getKey());
	// //p.add(key.getColumF(), key.getColumQ(), Bytes.toBytes(pageview));
	//
	// Increment incre = new Increment(key.getKey());
	// incre.addColumn(key.getColumF(), key.getColumQ(), pageview);
	//
	// context.write(null, incre);
	// }
	// }

	/**
	 * @uml.property  name="outPutFolder"
	 */
	public String outPutFolder ;
	
	
	@Override
	public int run(String[] arg0) throws Exception {
		// TODO Auto-generated method stub

		Job job = new Job(conf);
		job.setJarByClass(this.getClass());
		job.setJobName("Statistic-" + System.currentTimeMillis());

		job.setMapperClass(MapperClass.class);
		job.setReducerClass(ReducerClass.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setMapOutputKeyClass(Model.class);
		job.setMapOutputValueClass(IntWritable.class);

		// TableMapReduceUtil.initTableReducerJob("statistics", // output table
		// TableReducerClass.class, // reducer class
		// job);

		for (String s : arg0) {
			System.out.println("Add input path " + s);
			FileInputFormat.addInputPath(job, new Path(s));
		}

		// job.setNumReduceTasks(1);

		//
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		FileOutputFormat.setOutputPath(job, new Path(outPutFolder));
		job.setOutputKeyClass(Model.class);
		job.setOutputValueClass(LongWritable.class);
		job.setNumReduceTasks(6);
		job.waitForCompletion(true);
		return 0;
	}

	public static void main(String[] args) {
		Configuration conf = new Configuration();
		JobStatistic jobStatistic = new JobStatistic();
		try {
			ToolRunner.run(conf, jobStatistic, args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
